// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.utils.hbase;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.meta.TSMeta;
import net.opentsdb.meta.TSUIDQuery;
import net.opentsdb.meta.UIDMeta;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Assert;
import net.opentsdb.utils.enums.FamQuls;
import org.hbase.async.AtomicIncrementRequest;
import org.hbase.async.Bytes;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 连接hbase操作工具类
 * 项目上实在太多代码混合在一起，将部分hbase操作剥离
 *
 */
public class HbaseMetaUtils extends HbaseAbsUtils{
    private static final Logger LOG = LoggerFactory.getLogger(HbaseMetaUtils.class);

    /** The start row to scan on empty search strings.  `!' = first ASCII char. */
    private static final byte[] START_ROW = new byte[] { '!' };
    /** The end row to scan on empty search strings.  `~' = last ASCII char. */
    private static final byte[] END_ROW = new byte[] { '~' };
    /** tsdb meta 计数使用的 tsuid*/
    private static final String TSUID_COUNT_EKY = "00001000010000100001";

    public HbaseMetaUtils(final HBaseClient client, final byte[] table){
        super(client, table);
    }


    /**
     * 获取TSUID_COUNT_EKY计数的值
     *
     * @return
     */
    public Long getTsuidCount(){

        List<KeyValue> value = fetchData(UniqueId.stringToUid(TSUID_COUNT_EKY), null, null);

        Long count = null;

        if(value !=null && value.size() > 0){
            count = Bytes.getLong(value.get(0).value());
        }

//        LOG.debug("get the count: "+count);

        return count;
    }

    /**
     * @see #existsTsuid(byte[])
     * */
    public boolean existsDelTsuid(byte[] tsuid){

        return existsData(tsuid,FamQuls.META.FAMILY.DEL.getByte(),null);
    }

    /**
     * 判断对应的tsuid是否存在
     *
     * @param tsuid
     * @return
     */
    public boolean existsTsuid(byte[] tsuid){

        return existsData(tsuid,TSMeta.FAMILY(),FamQuls.META.QULITIFER.COUNTER.getByte());
    }

    /**
     * 判断meta是否存在metric和tagk对应的时间线，目前用作清理tag表使用
     *
     * @param metric
     * @return
     */
    public Set<String> fetchMetricAllTagk(TSDB tsdb, String metric){
        Set<String> sourceTagks = Sets.newConcurrentHashSet();
        // 将所有的时间线的tagk保存起来
        for(TSMeta meta : queryTsuid(tsdb, metric, Maps.newHashMap())){
            for(UIDMeta uid :meta.getTags()){
                if(uid.getType().equals(UniqueId.UniqueIdType.TAGK)){
                    sourceTagks.add(uid.getName());
                }
            }
        }

        return sourceTagks;
    }

    /**
     * 根据条件查询所有的时间线
     *
     * @param tsdb
     * @param metric
     * @param tags
     * @return
     */
    public List<TSMeta> queryTsuid(TSDB tsdb, String metric, Map<String, String> tags){
        final TSUIDQuery tsuidQuery = new TSUIDQuery(tsdb, metric, tags);
        List<TSMeta> tsMetas = Lists.newArrayList();
        try{
            tsMetas = tsuidQuery.getTSMetas()
                    .joinUninterruptibly();
        }catch (Exception e){
            LOG.error("query meta error:",e);
        }

        return tsMetas;
    }

    /**
     * @see #storeTsmetaData(byte[], FamQuls, FamQuls, long)
     * */
    public void initTSMetaCount(long n){
        storeTsmetaData(
                UniqueId.stringToUid(TSUID_COUNT_EKY),
                FamQuls.META.FAMILY.COUNT,
                FamQuls.META.QULITIFER.COUNTER,
                n<0?0:n);
    }

    /**
     * @see #storeTsmetaData(byte[], FamQuls, FamQuls, long)
     * */
    public void storeCountTSMeta(byte[] tsuid, long created){
        storeTsmetaData(tsuid, FamQuls.META.FAMILY.NAME, FamQuls.META.QULITIFER.COUNTER, created);
    }

    /**
     * @see #storeTsmetaData(byte[], FamQuls, FamQuls, long)
     * */
    public boolean storeCasCountTSMeta(byte[] tsuid, long created){

        return casStoreData(tsuid,
                FamQuls.META.FAMILY.NAME.getByte(),
                FamQuls.META.QULITIFER.COUNTER.getByte(),
                Bytes.fromLong(created),
                EMPTY_ARRAY);
    }

    /**
     * @see #storeTsmetaData(byte[], FamQuls, FamQuls, long)
     */
    public void storeDelTSMeta(byte[] tsuid){
        storeTsmetaData(tsuid, FamQuls.META.FAMILY.DEL, FamQuls.META.QULITIFER.COUNTER, 0);
    }

    /**
     * @see #casStoreData(byte[], byte[], byte[], byte[], byte[])
     */
    public boolean storeCasDelTSMeta(byte[] tsuid){
        return casStoreData(tsuid,
                FamQuls.META.FAMILY.DEL.getByte(),
                FamQuls.META.QULITIFER.COUNTER.getByte(),
                Bytes.fromLong(0L),
                EMPTY_ARRAY);
    }

    /**
     * 插入数据
     *
     * @param tsuid         tsuid
     * @param qulitifer     qulitifer name
     * @param num           created number
     */
    public void storeTsmetaData(byte[] tsuid,FamQuls family, FamQuls qulitifer, long num){
        storeData(tsuid, family.getByte() , qulitifer.getByte(), Bytes.fromLong(num));
    }


    /**
     * @see #bufferMetaAtomicRequest(byte[], FamQuls, long)
     * */
    public Deferred<Long> incrementDefaultAndGetCounter(){
        return bufferMetaAtomicRequest(UniqueId.stringToUid(TSUID_COUNT_EKY),
                FamQuls.META.QULITIFER.COUNTER, 1);
    }

    /**
     * @see #bufferMetaAtomicRequest(byte[], FamQuls, long)
     * */
    public Deferred<Long> decrementDefaultAndGetCounter(){
        return bufferMetaAtomicRequest(UniqueId.stringToUid(TSUID_COUNT_EKY),
                FamQuls.META.QULITIFER.COUNTER, -1);
    }

    /**
     * @see #bufferMetaAtomicRequest(byte[], FamQuls, long)
     * */
    public Deferred<Long> incrementAndGetCounter(final byte[] tsuid){
        return bufferMetaAtomicRequest(tsuid, FamQuls.META.QULITIFER.COUNTER, 1);
    }

    /**
     * @see #bufferMetaAtomicRequest(byte[], FamQuls, long)
     * */
    public Deferred<Long> decrementAndGetCounter(final byte[] tsuid){
        return bufferMetaAtomicRequest(tsuid, FamQuls.META.QULITIFER.COUNTER, -1);
    }

    /**
     * 自动发送buffer计数请求
     *
     * @param tsuid
     * @param qulitifer
     * @param num
     * @return
     */
    public Deferred<Long> bufferMetaAtomicRequest(
            final byte[] tsuid, FamQuls qulitifer,long num) {
        Assert.notNull(tsuid,"tsuid con't be null");

        final AtomicIncrementRequest inc = new AtomicIncrementRequest(
                table, tsuid, FamQuls.META.FAMILY.COUNT.getByte(),
                qulitifer.getByte(), num);

        return client.bufferAtomicIncrement(inc);
    }


    /**
     * @see #deleteMeta(byte[], FamQuls)
     * */
    public Deferred<Object> deleteCountMeta(final byte[] tsuid){
        return deleteMeta(tsuid, FamQuls.META.QULITIFER.COUNTER);
    }

    /**
     * @see #deleteMeta(byte[], FamQuls)
     * */
    public Deferred<Object> deleteMetaqulitiferMeta(final byte[] tsuid){
        return deleteMeta(tsuid, FamQuls.META.QULITIFER.META);
    }

    /**
     * 删除tsuid
     *
     * @param tsuid
     * @param qulitifer
     * @return
     */
    public Deferred<Object> deleteMeta(final byte[] tsuid,FamQuls qulitifer){
        return deleteData(tsuid, TSMeta.FAMILY(), qulitifer.getByte());
    }

    /**
     * 批量删除时间线
     *
     * @param tsuids
     * @throws Exception
     */
    public void deleteTsuids(final List<String> tsuids) throws Exception{
        final List<Deferred<Object>> delGroup = new ArrayList<>();

        tsuids.stream().forEach( tsuid ->{
            byte[] byteUid = UniqueId.stringToUid(tsuid);
            /*if(!existsDelTsuid(byteUid)
                    && existsTsuid(byteUid)){
                delGroup.add(deleteCountMeta(byteUid));
                if(storeCasDelTSMeta(byteUid))
                    decrementDefaultAndGetCounter();
            }*/

            // 通过del判断是否删除会造成无法重复删除问题,所以去除del判断
            if(existsTsuid(byteUid)){
                delGroup.add(deleteCountMeta(byteUid));
                decrementDefaultAndGetCounter();
            }
        });

        Deferred.group(delGroup).joinUninterruptibly();

        // 删除完成矫正计数器
        inCorrectCount();
    }

    /**
     * 矫正hbase计数
     *
     * @return
     */
    public long inCorrectCount(){
        ArrayList<ArrayList<KeyValue>> results =
                scanTableData(FamQuls.META.FAMILY.NAME.getByte());

        Long sum = 0L;
        if(results != null || results.size() > 0){
            for(List  list: results)
                sum += list.size();
        }

        initTSMetaCount(sum);

        return sum;
    }

}
